/////////////////////////////////////////////////////////////////////////////
// 2023 -
//open source under Apache License Version 2.0
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <kmhdfs/client/striped_input_stream_impl.h>
#include <kmhdfs/client/position_stripe_reader.h>
#include <kmhdfs/client/raw_erasure_decoder.h>
#include <kmhdfs/client/ec_chunk.h>
#include <kmhdfs/client/preconditions.h>
#include <turbo/log/logging.h>

namespace Hdfs {
namespace Internal {

class CorruptedBlocks;
PositionStripeReader::PositionStripeReader(StripedBlockUtil::AlignedStripe & alignedStripe,
                                           std::shared_ptr<ECPolicy> ecPolicy,
                                           std::vector<LocatedBlock> & targetBlocks,
                                           std::vector<StripeReader::BlockReaderInfo *> & readerInfos,
                                           std::shared_ptr<CorruptedBlocks> corruptedBlocks,
                                           std::shared_ptr<RawErasureDecoder> decoder,
                                           StripedInputStreamImpl * dfsStripedInputStream,
                                           std::shared_ptr<SessionConfig> conf) :
                                           StripeReader(alignedStripe, ecPolicy, targetBlocks,
                                                        readerInfos, corruptedBlocks, decoder,
                                                        dfsStripedInputStream, conf) {
}

PositionStripeReader::~PositionStripeReader() {
}

void PositionStripeReader::prepareDecodeInputs() {
    if (codingBuffer == nullptr) {
        decodeInputs = std::vector<std::shared_ptr<ECChunk>>(dataBlkNum + parityBlkNum);
        initDecodeInputs(alignedStripe);
    }
}

void PositionStripeReader::initDecodeInputs(StripedBlockUtil::AlignedStripe & alignedStripe) {
    int bufLen = (int) alignedStripe.getSpanInBlock();
    int bufCount = dataBlkNum + parityBlkNum;
    codingBuffer = std::shared_ptr<ByteBuffer>(new ByteBuffer(bufLen * bufCount));
    std::shared_ptr<ByteBuffer> buffer;
    for (int i = 0; i < dataBlkNum; i++) {
        buffer = std::shared_ptr<ByteBuffer>(codingBuffer->duplicate());
        decodeInputs[i] = std::shared_ptr<ECChunk>(new ECChunk(buffer, i * bufLen, bufLen));
    }

    for (int i = 0; i < dataBlkNum; i++) {
        if (alignedStripe.chunks[i] == nullptr) {
            alignedStripe.chunks[i] =
                    std::shared_ptr<StripedBlockUtil::StripingChunk>(new StripedBlockUtil::StripingChunk(decodeInputs[i]->getBuffer()));
        }
    }
}

bool PositionStripeReader::prepareParityChunk(int index) {
    Preconditions::checkState(index >= dataBlkNum &&
                              alignedStripe.chunks[index] == nullptr);

    int bufLen = (int) alignedStripe.getSpanInBlock();
    decodeInputs[index] = std::shared_ptr<ECChunk>(new ECChunk(std::shared_ptr<ByteBuffer>(codingBuffer->duplicate()),
                                                                                 index * bufLen, bufLen));

    alignedStripe.chunks[index] =
            std::shared_ptr<StripedBlockUtil::StripingChunk>(new StripedBlockUtil::StripingChunk(decodeInputs[index]->getBuffer()));

    return true;
}

void PositionStripeReader::decode() {
    VKLOG(300)<<turbo::str_format( "decoding!!!");
    finalizeDecodeInputs();
    decodeAndFillBuffer(true);
}

/**
 * Default close do nothing.
 */
void PositionStripeReader::close() {
}

}
}
